package com.study.rabbitmq.mqtt2;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;

@Slf4j
@Configuration
public class IotMqttSubscriberConfig {

    private static final String INPUT_CHANNEL = "iotMqttSubInputChannel";

    @Autowired
    private MqttTwoConfig mqttConfig;

    @Bean
    public MqttPahoClientFactory mqttClientSubFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setServerURIs(mqttConfig.getServers());
        return factory;
    }

    @Bean
    public MessageChannel iotMqttSubInputChannel() {
        return new DirectChannel();
    }

    @Bean
    public MessageProducer subInbound(@Qualifier("mqttAdapterTwo") MqttPahoMessageDrivenChannelAdapter adapter) {
//        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(mqttConfig.getClientId(), mqttClientSubFactory(), mqttConfig.getDefaultTopic());
        adapter.setCompletionTimeout(5000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(1);
        adapter.setOutputChannel(iotMqttSubInputChannel());
        return adapter;
    }

    @Bean(name="mqttAdapterTwo")
    public MqttPahoMessageDrivenChannelAdapter adapter(){
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(mqttConfig.getClientId(), mqttClientSubFactory(), mqttConfig.getDefaultTopic());
        return adapter;
    }

    /**
     *  消息订阅
     */
    @Bean
    @ServiceActivator(inputChannel = INPUT_CHANNEL)
    public MessageHandler handlerTest() {

        return message -> {
            try {
                String string = message.getPayload().toString();
                log.info("接收到消息：" + string);
            } catch (MessagingException ex) {
                log.error("接收消息发生异常：{}", ex.getMessage());
            }
        };
    }
}
